package com.econ.powercloud.ems.kafka;

import com.econ.powercloud.common.base.facade.util.JacksonUtil;
import com.econ.powercloud.datahandle.facade.domain.ProcessedCalData;
import com.econ.powercloud.datahandle.facade.domain.ProcessedRunData;
import com.econ.powercloud.ems.common.enums.DitTypeEnum;
import com.econ.powercloud.ems.common.enums.MeasureType;
import com.econ.powercloud.ems.common.interf.AbstractEconBusProcess;
import com.econ.powercloud.ems.common.pojo.*;
import com.econ.powercloud.ems.common.interf.*;
import com.econ.powercloud.ems.kafka.strategy.EconCommonBusStrategy;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.core.annotation.AnnotationAwareOrderComparator;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;
import org.springframework.util.CollectionUtils;

import java.util.List;
import java.util.stream.Collectors;

/**
 * @Author : zjf
 * @Description : kafka消费者
 * @Date : 2022/7/14 10:30
 * @Modified by : zjf
 */
@Slf4j
@Component
public class DataHandlerDataConsumer {

    @Autowired
    private List<AnalogDataObserver> analogDataObserverList;
    @Autowired
    private List<StatusDataObserver> statusDataObserverList;
    @Autowired
    private List<CalAnalogDataObserver> calAnalogDataObserverList;
    @Autowired
    private List<CalStatusDataObserver> calStatusDataObserverList;
    @Autowired
    private List<EventDataObserver> eventDataObserverList;
    @Autowired
    private List<ProtectSoeDataObserver> protectSoeDataObserverList;
    @Autowired
    private List<StatusAlarmFaultDataObserver> statusAlarmFaultDataObserverList;
    @Autowired(required = false)
    private List<ServiceFeedbackDataObserver> observerList;
    @Autowired
    private List<WarnDataObserver> warnDataObserverList;
    @Autowired
    private EmsYcYxStrategyObserver emsYcYxStrategyObserver;
    @Autowired
    private EconCommonBusStrategy econCommonBusStrategy;



    /**
     * 遥测遥信websocket数据推送
     *
     * @param record
     */
    @KafkaListener(topics = "${dataflow.consumer.run-data-topic}", groupId = "${ems.run-data.groupId}")
    public void rundataReceive(ConsumerRecord<String, String> record) {
        String value = record.value();
        if (StringUtils.isBlank(value)) {
            return;
        }
        List<ProcessedRunData> values = JacksonUtil.parseList(value, ProcessedRunData.class);
        if (values.size() == 0) {
            return;
        }
        //按照type进行分类
        List<ProcessedRunData> yxRunDataList = values.stream().filter(d -> MeasureType.yx.getCodeInIot().equals(d.getType())).collect(Collectors.toList());
        List<ProcessedRunData> ycRunDataList = values.stream().filter(d -> MeasureType.yc.getCodeInIot().equals(d.getType())).collect(Collectors.toList());
        if (!CollectionUtils.isEmpty(ycRunDataList)) {
            List<EmsKafkaYcDataDTO> ycDataList = JacksonUtil.convertToList(ycRunDataList, EmsKafkaYcDataDTO.class);
            ycDataList.stream().forEach(data -> {
                data.setDit(data.getPid().replace(data.getDeviceCode() + ".", ""));
                data.setDitType(DitTypeEnum.YC.getCode());
            });
            for (AnalogDataObserver analogDataObserver : analogDataObserverList) {
                analogDataObserver.doInvoke(ycDataList);
            }
        }
        if (!CollectionUtils.isEmpty(yxRunDataList)) {
            List<EmsKafkaYxDataDTO> yxDataList = JacksonUtil.convertToList(yxRunDataList, EmsKafkaYxDataDTO.class);
            yxDataList.stream().forEach(data -> {
                data.setDit(data.getPid().replace(data.getDeviceCode() + ".", ""));
                data.setDitType(DitTypeEnum.YX.getCode());
            });
            for (StatusDataObserver statusDataObserver : statusDataObserverList) {
                yxDataList.stream().forEach(d -> {
                    statusDataObserver.doInvoke(d);
                });
            }
        }
    }

    /**
     * 计算数据websocket推送
     *
     * @param record
     */
    @KafkaListener(topics = "${dataflow.consumer.cal-data-topic}", groupId = "${ems.cal-data.groupId}")
    public void calDataReceive(ConsumerRecord<String, String> record) {
        String value = record.value();
        if (StringUtils.isBlank(value)) {
            return;
        }
        List<ProcessedCalData> values = JacksonUtil.parseList(value, ProcessedCalData.class);
        if (values.size() == 0) {
            return;
        }
        //按照type进行分类
        List<ProcessedCalData> yxRunDataList = values.stream().filter(d -> MeasureType.yx.getCodeInIot().equals(d.getType())).collect(Collectors.toList());
        List<ProcessedCalData> ycRunDataList = values.stream().filter(d -> MeasureType.yc.getCodeInIot().equals(d.getType())).collect(Collectors.toList());
        if (!CollectionUtils.isEmpty(ycRunDataList)) {
            List<EmsKafkaYcDataDTO> ycDataList = JacksonUtil.convertToList(ycRunDataList, EmsKafkaYcDataDTO.class);
            ycDataList.stream().forEach(data -> {
                data.setDit(data.getPid().replace(data.getDeviceCode() + ".", ""));
                data.setDitType(DitTypeEnum.CAL_YC.getCode());
            });
            for (CalAnalogDataObserver calAnalogDataObserver : calAnalogDataObserverList) {
                ycDataList.stream().forEach(d -> {
                    calAnalogDataObserver.doInvoke(d);
                });
            }
        }
        if (!CollectionUtils.isEmpty(yxRunDataList)) {
            List<EmsKafkaYxDataDTO> yxDataList = JacksonUtil.convertToList(yxRunDataList, EmsKafkaYxDataDTO.class);
            yxDataList.stream().forEach(data -> {
                data.setDit(data.getPid().replace(data.getDeviceCode() + ".", ""));
                data.setDitType(DitTypeEnum.CAL_YX.getCode());
            });
            for (CalStatusDataObserver calStatusDataObserver : calStatusDataObserverList) {
                yxDataList.stream().forEach(d -> {
                    calStatusDataObserver.doInvoke(d);
                });
            }
        }
    }

    /**
     * 事件->故障告警记录处理
     *
     * @param record
     */
    @KafkaListener(topics = "${dataflow.consumer.event-data-topic}", groupId = "ems-event-kafka")
    public void eventReceive(ConsumerRecord<String, String> record) {
        String value = record.value();
        if (StringUtils.isBlank(value)) {
            return;
        }
        List<EmsKafkaEvenDataDTO> eventDataList = JacksonUtil.parseList(value, EmsKafkaEvenDataDTO.class);
        if (eventDataList.size() == 0) {
            return;
        }
        for (EventDataObserver eventDataObserver : eventDataObserverList) {
            eventDataList.stream().forEach(d -> eventDataObserver.doInvoke(d));
        }
    }

    /**
     * 事件->保护/SOE事件处理
     *
     * @param record
     */
    @KafkaListener(topics = "${dataflow.consumer.event-data-topic}", groupId = "ems-event-protect-soe-kafka")
    public void protectSoeEventReceive(ConsumerRecord<String, String> record) {
        String value = record.value();
        if (StringUtils.isBlank(value)) {
            return;
        }
        List<EmsKafkaEvenDataDTO> eventDataList = JacksonUtil.parseList(value, EmsKafkaEvenDataDTO.class);
        if (eventDataList.size() == 0) {
            return;
        }
        for (ProtectSoeDataObserver protectSoeDataObserver : protectSoeDataObserverList) {
            eventDataList.stream().forEach(d -> protectSoeDataObserver.doInvoke(d));
        }
    }

    /**
     * 遥信的故障告警的处理
     *
     * @param record
     */
    @KafkaListener(topics = "${dataflow.consumer.run-data-topic}", groupId = "ems-rundata-kafka-yx-fault")
    public void yxFaultReceive(ConsumerRecord<String, String> record) {
        String value = record.value();
        if (StringUtils.isBlank(value)) {
            return;
        }
        List<ProcessedRunData> values = JacksonUtil.parseList(value, ProcessedRunData.class);
        if (values.size() == 0) {
            return;
        }
        //只有遥信数据才可能产生故障告警
        List<ProcessedRunData> yxRunDataList = values.stream().filter(d -> MeasureType.yx.getCodeInIot().equals(d.getType())).collect(Collectors.toList());
        if (!CollectionUtils.isEmpty(yxRunDataList)) {
            List<EmsKafkaYxDataDTO> yxDataList = JacksonUtil.convertToList(yxRunDataList, EmsKafkaYxDataDTO.class);
            for (StatusAlarmFaultDataObserver statusAlarmFaultDataObserver : statusAlarmFaultDataObserverList) {
                statusAlarmFaultDataObserver.doInvoke(yxDataList);
            }
        }
        //处理yk/yt下发值(置0)
        if (yxRunDataList.size() > 0) {
            emsYcYxStrategyObserver.doInvoke(yxRunDataList);
        }
    }

    @KafkaListener(topics = "${dataflow.consumer.device-reply-topic}", groupId = "ems-device-reply-kafka")
    public void replyReceive(ConsumerRecord<String, String> record) {
        String value = record.value();
        if (StringUtils.isBlank(value)) {
            return;
        }
        EmsMqttReplyVO replyVO = JacksonUtil.parseObject(record.value(), EmsMqttReplyVO.class);
        for (ServiceFeedbackDataObserver observer : observerList) {
            observer.doInvoke(replyVO);
        }
    }

    /**
     * 遥测-诊断预警处理
     *
     * @param record
     */
    @KafkaListener(topics = "${dataflow.consumer.run-data-topic}", groupId = "ems-warn-yc-kafka")
    public void warnYcReceive(ConsumerRecord<String, String> record) {
        String value = record.value();
        if (StringUtils.isBlank(value)) {
            return;
        }
        List<ProcessedRunData> values = JacksonUtil.parseList(value, ProcessedRunData.class);
        if (values.size() == 0) {
            return;
        }
        //按照type进行分类
        List<ProcessedRunData> ycRunDataList = values.stream().filter(d -> MeasureType.yc.getCodeInIot().equals(d.getType())).collect(Collectors.toList());
        if (!CollectionUtils.isEmpty(ycRunDataList)) {
            List<ProcessedDataDTO> ycDataList = JacksonUtil.convertToList(ycRunDataList, ProcessedDataDTO.class);
            ycDataList.stream().forEach(d -> d.setDitType(DitTypeEnum.YC.getCode()));
            for (WarnDataObserver warnDataObserver : warnDataObserverList) {
                ycDataList.stream().forEach(d -> {
                    warnDataObserver.doInvoke(d);
                });
            }
        }
    }

    /**
     * 计算遥测-诊断预警处理
     *
     * @param record
     */
    @KafkaListener(topics = "${dataflow.consumer.cal-data-topic}", groupId = "ems-warn-cal-yc-kafka")
    public void warnCalYcReceive(ConsumerRecord<String, String> record) {
        String value = record.value();
        if (StringUtils.isBlank(value)) {
            return;
        }
        List<ProcessedCalData> values = JacksonUtil.parseList(value, ProcessedCalData.class);
        if (values.size() == 0) {
            return;
        }
        //按照type进行分类
        List<ProcessedCalData> ycRunDataList = values.stream().filter(d -> MeasureType.yc.getCodeInIot().equals(d.getType())).collect(Collectors.toList());
        if (!CollectionUtils.isEmpty(ycRunDataList)) {
            List<ProcessedDataDTO> ycDataList = JacksonUtil.convertToList(ycRunDataList, ProcessedDataDTO.class);
            ycDataList.stream().forEach(d -> d.setDitType(DitTypeEnum.CAL_YC.getCode()));
            for (WarnDataObserver warnDataObserver : warnDataObserverList) {
                ycDataList.stream().forEach(d -> {
                    warnDataObserver.doInvoke(d);
                });
            }
        }
    }

    /**
     * 推送内存同步刷新信息到kafka
     *
     * @param record
     */
    @KafkaListener(topics = "${dataflow.producer.econ-common-bus-topic}", groupId = "${ems.econ-common-bus.groupId}")
    public void memoryDataSyncRefresh(ConsumerRecord<String, String> record) {
        String value = record.value();
        if (StringUtils.isBlank(value)) {
            return;
        }
        List<CacheOperationVO> operationVOList = JacksonUtil.parseList(value, CacheOperationVO.class);
        if(CollectionUtils.isEmpty(operationVOList)){
            return;
        }
        for (CacheOperationVO cacheOperationVO : operationVOList){
            econCommonBusStrategy.operationCache(cacheOperationVO);
        }
    }

}
